package cn.dmrliu.edu.realtime.app.dws;

import cn.dmrliu.edu.realtime.bean.ExamQuestionBean;
import cn.dmrliu.edu.realtime.util.ClickhouseUtil;
import cn.dmrliu.edu.realtime.util.DateUtil;
import cn.dmrliu.edu.realtime.util.KafkaUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsToolExamQuestionWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1. 基本环境的准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // TODO 2. 检查点的设置

        // TODO 3.从kafka的 exam 主题读取数据
        String topic = "edu_dwd_tool_exam";
        String group = "edu_dws_tool_exam_question_group";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(topic, group);
        DataStreamSource<String> kafkaDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka source");

        // TODO 4.转为可计算的实体类，配置计算逻辑:答题次数，正确答题次数
        SingleOutputStreamOperator<ExamQuestionBean> beanDS = kafkaDS.map(new MapFunction<String, ExamQuestionBean>() {
            @Override
            public ExamQuestionBean map(String value) throws Exception {

                JSONObject jsonObject = JSONObject.parseObject(value);

                String questionId = jsonObject.getString("question_id");
                String userId = jsonObject.getString("user_id");

                String submitTime = jsonObject.getString("submit_time");
                Long ts = DateUtil.toTs(submitTime, true);

                Long corCt = 0L;
                Boolean correctFlag = false;
                String isCorrect = jsonObject.getString("is_correct");
                if ("1".equals(isCorrect)) {
                    corCt = 1L;
                    correctFlag = true;
                }
                return new ExamQuestionBean(
                        "", "",
                        questionId,
                        userId,
                        correctFlag,
                        corCt,
                        1L,
                        0L,
                        0L,
                        ts
                );

            }
        });

        // TODO 6.以 题目 和 用户 为组  ， 配置计算逻辑: 答题次数独立用户 ，正确答题次数独立用户
        SingleOutputStreamOperator<ExamQuestionBean> examQuestionDS = beanDS
                .keyBy(new KeySelector<ExamQuestionBean, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, String> getKey(ExamQuestionBean value) throws Exception {
                        return Tuple2.of(value.getQuestionId(), value.getUserId());
                    }
                })
                .process(new ProcessFunction<ExamQuestionBean, ExamQuestionBean>() {

                    private ValueState<Boolean> isNewState;// 记录是否是新独立用户
                    private ValueState<Boolean> isCorrectState;// 记录是否正确回答


                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<Boolean> isNewVSD = new ValueStateDescriptor<>(
                                "is_new_user", Boolean.class
                        );
                        isNewVSD.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        isNewState = getRuntimeContext().getState(isNewVSD);

                        ValueStateDescriptor<Boolean> isCorrectVSD = new ValueStateDescriptor<>(
                                "hava_correct", Boolean.class
                        );
                        isCorrectVSD.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        isCorrectState = getRuntimeContext().getState(isCorrectVSD);
                    }

                    @Override
                    public void processElement(ExamQuestionBean value, Context ctx, Collector<ExamQuestionBean> out) throws Exception {
                        Boolean isNew = isNewState.value();

                        // 是否是新独立用户
                        if (isNew == null) {
                            value.setUAnsCt(1L);
                            isNewState.update(false);
                        }

                        // 回答正确，且还没有计数过就 +1
                        Boolean hadCorrect = isCorrectState.value();

                        if (value.getIsCorrect() && hadCorrect == null) {
                            value.setUCorCt(1L);
                            isCorrectState.update(false);
                        }

                        out.collect(value);
                    }
                });


        // TODO 7.设置水位线
        SingleOutputStreamOperator<ExamQuestionBean> watermarkDS = examQuestionDS
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<ExamQuestionBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<ExamQuestionBean>() {
                                    @Override
                                    public long extractTimestamp(ExamQuestionBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                })
                );

        // TODO 8.以题目分组，开窗，聚合
        SingleOutputStreamOperator<ExamQuestionBean> reduceDS = watermarkDS
                .keyBy(in -> in.getQuestionId())
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(new ReduceFunction<ExamQuestionBean>() {
                            @Override
                            public ExamQuestionBean reduce(ExamQuestionBean value1, ExamQuestionBean value2) throws Exception {
                                value1.setCorCt(value1.getCorCt() + value2.getCorCt());
                                value1.setAnsCt(value1.getAnsCt() + value2.getAnsCt());
                                value1.setUCorCt(value1.getUCorCt() + value2.getUCorCt());
                                value1.setUAnsCt(value1.getUAnsCt() + value2.getUAnsCt());
                                return value1;
                            }
                        },
                        new WindowFunction<ExamQuestionBean, ExamQuestionBean, String, TimeWindow>() {
                            @Override
                            public void apply(String key, TimeWindow window, Iterable<ExamQuestionBean> input, Collector<ExamQuestionBean> out) throws Exception {
                                String stt = DateUtil.toYmdHms(window.getStart());
                                String edt = DateUtil.toYmdHms(window.getEnd());
                                for (ExamQuestionBean examQuestionBean : input) {
                                    examQuestionBean.setStt(stt);
                                    examQuestionBean.setEdt(edt);
                                    out.collect(examQuestionBean);
                                }
                            }
                        });

        reduceDS.print(">>>>>>>>>>");

        // TODO 9.写入clickhouse
        reduceDS.addSink(ClickhouseUtil.getSinkFunction("insert into dws_tool_exam_question_window values(?,?,?,?,?,?,?,?)"));

        env.execute();

    }
}
